package cn.edu.xmu.datamining.tangzk.mralgos;

import java.io.IOException;
import java.util.Arrays;
import java.util.Date;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ProgramDriver;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


import cn.edu.xmu.datamining.tangzk.mralgos.kmeans.KMeansDriver;
import cn.edu.xmu.datamining.tangzk.mralgos.linreg.MultipleLinearRegressionMapReduce;
import cn.edu.xmu.datamining.tangzk.mralgos.pagerank.PageRankDriver;
import cn.edu.xmu.datamining.tangzk.mralgos.parallelbfs.ParalleBFSDriver;
import cn.edu.xmu.datamining.tangzk.mralgos.secondarysort.SecondarySortDriver;
import cn.edu.xmu.datamining.tangzk.mralgos.sssp.SingleSourceShortestPathDriver;
import cn.edu.xmu.datamining.tangzk.util.ConvertCsvToSeqDriver;
import cn.edu.xmu.datamining.tangzk.util.Joiner;

public class AlgorithmDriver {

	public static void main(String[] args) {
		int exitCode = -1;
		ProgramDriver pgd = new ProgramDriver();
		try {
			pgd.addClass("pbfs", ParalleBFSDriver.class,
					"parallel breadth first search");
			pgd.addClass("sesort", SecondarySortDriver.class, "secondary sort.");
			pgd.addClass("sssp", SingleSourceShortestPathDriver.class,
					"single source shortest path.");
			pgd.addClass("pagerank", PageRankDriver.class,
					"pagerank calaculating.");
			pgd.addClass("kmeans", KMeansDriver.class, "KMeans clustering.");
			pgd.addClass("seq", ConvertCsvToSeqDriver.class,
					"Convert CSV to SquenceFile.");
			pgd.addClass("linreg", MultipleLinearRegressionMapReduce.class,
					"Linear Regression");
			pgd.driver(args);
			// Success
			exitCode = 0;
		} catch (Throwable e) {
			e.printStackTrace();
		}
		System.exit(exitCode);
	}

	public static void toolRun(Tool driver, String[] args) throws Exception {
		ToolRunner.run(driver, args);
	}

	public static int runJob(Job job) throws ClassNotFoundException,
			IOException, InterruptedException {
		return runJob(job, true);
	}

	public static int runJob(Job job, boolean verbose) throws IOException,
			InterruptedException, ClassNotFoundException {
		String jobName = job.getJobName();
		job.setJarByClass(AlgorithmDriver.class);
		printJobInfo(job);
		//
		// run job
		//
		Date startTime = new Date();
		System.out.println("{" + jobName + "} Job started: " + startTime);
		int flag = job.waitForCompletion(verbose) ? 0 : -1;
		Date end_time = new Date();
		System.out.println("{" + jobName + "} Job ended: " + end_time);
		System.out.println("The job took "
				+ (end_time.getTime() - startTime.getTime()) / (float) 1000.0
				+ " seconds.");
		return flag;
	}

	public static void printJobInfo(Job job) throws ClassNotFoundException {
		System.out.println("MapperClass:\t"
				+ job.getMapperClass().getSimpleName());
		System.out.println("ReducerClass:\t"
				+ job.getReducerClass().getSimpleName());
		System.out.println("NumReduceTasks:\t" + job.getNumReduceTasks());
		System.out.println("Input Path:\t"
				+ Joiner.on(Arrays.asList(FileInputFormat.getInputPaths(job))
						.iterator(), ","));
		System.out.println("Output Path:\t"
				+ FileOutputFormat.getOutputPath(job).toString());
	}
}
